package com.gis.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{DataFrame, Encoder, Encoders, Row, SparkSession, functions}


/**
 * @author LnnuUser
 * @create 2021-09-02-下午8:16
 */
object Spark03_SparkSQL_UDAF1 {

  def main(args: Array[String]): Unit = {

    //  TODO 创建SparkSQL的运行环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    val df: DataFrame = spark.read.json("datas/user.json")
    df.createOrReplaceTempView("user")

    spark.udf.register("ageAvg", functions.udaf(new MyAcgUDAF))

    spark.sql("select ageAvg(age) from user").show



    // TODO 执行逻辑操作


    // TODO 关闭环境
    spark.close()

  }

  /*
  自定义聚合函数类：计算年龄的平均值
  1.继承org.apache.spark.sql.expressions.Aggregator,定义泛型
    IN:输入的数据类型 Long
    BUF:缓冲区的数据类型 Buff
    OUT:输出的数据类型 Long
  2.重写方法(6个)
   */
  case class Buff(var total: Long, var count: Long)
  class MyAcgUDAF extends Aggregator[Long, Buff, Long] {

    // z&zero：初始值或者0值，缓冲区的初始化
    override def zero: Buff = {
      Buff(0L, 0L)
    }

    // 根据输入的数据更新缓冲区
    override def reduce(buff: Buff, in: Long): Buff = {
      buff.total = buff.total + in
      buff.count = buff.count + 1
      buff
    }

    // 合并缓冲区
    override def merge(b1: Buff, b2: Buff): Buff = {
      b1.total = b1.total + b2.total
      b1.count = b1.count + b2.count
      b1
    }

    // 计算结果
    override def finish(reduction: Buff): Long = {
      reduction.total / reduction.count
    }

    // 缓冲区的编码操作
    override def bufferEncoder: Encoder[Buff] = Encoders.product

    // 输出的编码操作
    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }


}
